"""
Case Type   : pg_create_logical_replication_slot
Case Name   : 使用pg_create_logical_replication_slot创建逻辑复制槽，restart_lsn/confirmed_flush参数校验
Create At   : 2024/8/15
Owner       : @li-xin12345
Description :
    1、修改参数
    2、配置白名单
    3、使用pg_create_logical_replication_slot创建逻辑复制槽，仅指定restart_lsn/confirmed_flush
    4、使用pg_create_logical_replication_slot创建逻辑复制槽，同时指定restart_lsn/confirmed_flush
    5、清理环境
Expect      :
    1、成功
    2、成功
    3、合理报错
    4、参数有效时创建成功，无效时合理报错
    5、成功
History     :
"""

import os
import time
import unittest

from yat.test import Node
from yat.test import macro
from testcase.utils.Common import Common
from testcase.utils.CommonSH import CommonSH
from testcase.utils.Constant import Constant
from testcase.utils.Logger import Logger

pri_sh = CommonSH('PrimaryDbUser')
node_num = pri_sh.get_node_num()


@unittest.skipIf(3 > node_num, '不满足一主两备环境跳过')
class RowCompressTest(unittest.TestCase):
    def setUp(self):
        self.log = Logger()
        self.log.info(f'-----{os.path.basename(__file__)} start-----')
        self.common = Common()
        self.constant = Constant()
        self.std1_sh = CommonSH('Standby1DbUser')
        self.std2_sh = CommonSH('Standby2DbUser')
        self.pri_user_node = Node('PrimaryDbUser')
        self.std1_user_node = Node('Standby1DbUser')
        self.db_name = 'db_gs_retrieve_0043'
        self.tb_name = 't_gs_retrieve_0043'
        self.role = 'u_gs_retrieve_0043'
        self.slot_names = 'slot_gs_retrieve_0043'
        self.paras = {'autovacuum': 'on', 'synchronous_commit': 'off',
                      'wal_level': 'logical', 'enable_slot_log': 'on',
                      'max_replication_slots': 15}
        self.default_params = {'autovacuum': 'on', 'synchronous_commit': 'on',
                               'wal_level': 'hot_standby',
                               'enable_slot_log': 'off',
                               'max_replication_slots': 8}
        self.pg_hba_path = os.path.join(macro.DB_INSTANCE_PATH,
                                        macro.PG_HBA_FILE_NAME)
        self.pg_hba_list = [f'local replication {self.role}  trust',
                            f'host replication {self.role} 127.0.0.1/32 trust',
                            f'host replication {self.role} ::1/128 trust']

    def test_row_compress(self):
        text = '-----step1:修改参数;expect:成功-----'
        self.log.info(text)
        for key, value in self.paras.items():
            res = self.common.show_param(key)
            self.log.info(key)
            if res != value:
                res = pri_sh.execute_gsguc('set',
                                           self.constant.GSGUC_SUCCESS_MSG,
                                           f'{key}={value}')
                self.log.info(res)
                self.log.info('步骤2重启')
                self.assertTrue(res)

        text = "-----step2:备份pg_hba.conf, 配置白名单;expect:成功-----"
        self.log.info(text)
        cmd = f"\\cp {self.pg_hba_path} {self.pg_hba_path}_bak.bak"
        res = self.pri_user_node.sh(cmd).result()
        self.log.info(res)
        self.assertNotIn('cannot', res, f'执行失败: {text}')

        sql = f"drop role if exists {self.role};" \
              f"create role {self.role} with login password " \
              f"'{macro.COMMON_PASSWD}';" \
              f"alter role {self.role} with replication sysadmin;"
        res = pri_sh.execut_db_sql(sql)
        self.assertIn(self.constant.DROP_ROLE_SUCCESS_MSG, res,
                      f'执行失败: {text}')
        self.assertIn(self.constant.CREATE_ROLE_SUCCESS_MSG, res,
                      f'执行失败: {text}')
        self.assertIn(self.constant.ALTER_ROLE_SUCCESS_MSG, res,
                      f'执行失败: {text}')

        for i in self.pg_hba_list:
            res = pri_sh.execute_gsguc('reload',
                                       self.constant.GSGUC_SUCCESS_MSG,
                                       '', pghba_param=i)
            self.log.info(res)
            self.assertTrue(res)
        status = pri_sh.get_db_cluster_status('detail')
        self.log.info(status)
        res = pri_sh.restart_db_cluster()
        self.log.info(res)
        time.sleep(2)
        status = pri_sh.get_db_cluster_status('detail')
        self.log.info(status)
        self.assertTrue("Degraded" in status or "Normal" in status)

        text = '-----step3:使用pg_create_logical_replication_slot创建逻辑复制槽，'\
               '仅指定restart_lsn/confirmed_flush;expect:合理报错-----'
        self.log.info(text)
        res = self.common.show_param('wal_level')
        self.log.info('xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx')
        self.log.info(res)

        sql = f"select pg_create_logical_replication_slot(" \
              f"'{self.slot_names}_01', 'mppdb_decoding');"
        res = pri_sh.execut_db_sql(sql)
        self.log.info(res)
        get_lsn1 = res.splitlines()[-2].split(',')[-1].strip(')')
        self.log.info(get_lsn1)
        sql = f"select pg_create_logical_replication_slot(" \
              f"'{self.slot_names}_02', 'mppdb_decoding');"
        res = pri_sh.execut_db_sql(sql)
        self.log.info(res)
        get_lsn2 = res.splitlines()[-2].split(',')[-1].strip(')')
        self.log.info(get_lsn2)
        sql = f"select pg_create_logical_replication_slot(" \
              f"'{self.slot_names}_03', 'mppdb_decoding', '{get_lsn1}');"
        res = pri_sh.execut_db_sql(sql)
        self.log.info(res)
        self.assertIn('ERROR', res, f'执行失败:{text}')

        text = '-----step4:使用pg_create_logical_replication_slot创建逻辑复制槽，' \
               '同时指定restart_lsn/confirmed_flush;expect:合理报错-----'
        self.log.info(text)
        sql = f'''select pg_create_logical_replication_slot(
        '{self.slot_names}_04', 'mppdb_decoding', '{get_lsn1}', '{get_lsn2}');
        select pg_create_logical_replication_slot(
        '{self.slot_names}_05', 'mppdb_decoding', '{get_lsn2}', '{get_lsn1}');
        '''
        res = pri_sh.execut_db_sql(sql)
        self.log.info(res)
        self.assertNotIn('ERROR', res, f'执行失败:{text}')

        sql = f'''select pg_create_logical_replication_slot(
        '{self.slot_names}_06', 'mppdb_decoding', '{get_lsn2}', '123');
        select pg_create_logical_replication_slot(
        '{self.slot_names}_07', 'mppdb_decoding', '456', '123');
        select pg_create_logical_replication_slot(
        '{self.slot_names}_08', 'mppdb_decoding', '456', '{get_lsn1}');
        '''
        res = pri_sh.execut_db_sql(sql)
        self.log.info(res)
        self.assertIn('ERROR', res, f'执行失败:{text}')

    def tearDown(self):
        text = '-----step5:清理环境; expect:成功-----'
        self.log.info(text)
        self.log.info('--恢复pg_hba.conf--')
        cmd = f"if [ -e {self.pg_hba_path}_bak.bak ]; " \
              f"then \\cp {self.pg_hba_path}_bak.bak {self.pg_hba_path}; " \
              f"rm -rf {self.pg_hba_path}_bak.bak; fi"
        self.log.info(cmd)
        res = self.pri_user_node.sh(cmd).result()
        self.log.info(res)
        self.log.info('--恢复参数--')
        for key, value in self.default_params.items():
            res = self.common.show_param(key)
            self.log.info(key)
            if res != value:
                res = pri_sh.execute_gsguc('set',
                                           self.constant.GSGUC_SUCCESS_MSG,
                                           f'{key}={value}')
                self.log.info(res)
        res = pri_sh.restart_db_cluster()
        self.log.info(res)
        time.sleep(2)
        status = pri_sh.get_db_cluster_status('detail')
        self.log.info(status)
        for i in range(1, 9):
            sql = f"select pg_drop_replication_slot('{self.slot_names}_0{i}');"
            res = pri_sh.execut_db_sql(sql)
            self.log.info(res)
        self.assertTrue("Degraded" in status or "Normal" in status)
        self.log.info(f'-----{os.path.basename(__file__)} end-----')
